package bsp.diff;

import java.io.IOException;
import java.lang.reflect.InvocationTargetException;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hama.bsp.BSPPeer;
import org.apache.hama.bsp.sync.SyncException;

import util.DiffComputation;
import util.DiffMessage2;
import util.Pair;
import util.Parsable;
import util.ScanDList;
import util.Utilties;

public class DiffStepThree<T0 extends Parsable<T0>, T1 extends Parsable<T1>, T2 extends Parsable<T2>>
		extends DiffMapper<T0, T1, T2, DiffMessage2<T1,T2>> {

	T2 accSum;
	T1 accTims;
	int beginFlg;

	T0 insT0;
	T2 pacc;
	Utilties<T1> tl1;
	
	Utilties<T2> tl2;

	// private OpScan<T> plus;

	public DiffStepThree(DiffComputation<T0, T1, T2> di) {
		super(di);

		accTims = otimes.id();
		accSum = oplus.id();
		pacc = null;

		insT0 = p.getInstanceOfT0();
		tl1 = new Utilties<T1>();
		tl2 = new Utilties<T2>();

		beginFlg =0;
	}

	@Override
	public void map(
			BSPPeer<LongWritable, Text, Text, NullWritable, Text> contex)
			throws IOException, SyncException, InterruptedException {

		if(myId < numPeers-1) return;
		
		if(beginFlg ==0){
			chunkey = contex.getConfiguration().getLong("chunkey", 0);  
			beginFlg++;
		}
 
 		ScanDList<T2> psums = new ScanDList<T2>();
 		ScanDList<T1> ptimes = new ScanDList<T1>();
//
 		
//
//		// get all messages
     	int countMsg = 0;
     	Text msg = contex.getCurrentMessage();
     	System.out.println("get DiffMessage2: " + msg);
     	
     	DiffMessage2<T1, T2> receiver = new DiffMessage2<T1, T2>(msg.toString(), accTims.getClass(), accSum.getClass() );
		while ( msg != null) {

			countMsg++;
			Pair<T1, T2> previousAcc = new Pair<T1, T2>();
			try {
				previousAcc = receiver.parse(msg.toString()).getSecond(); // @@@
			} catch (NoSuchMethodException | SecurityException
					| ClassNotFoundException | InstantiationException
					| IllegalAccessException | IllegalArgumentException
					| InvocationTargetException e) {
				e.printStackTrace();
			}
			System.out.println("DiffMessage2: " + previousAcc.getFirst());
			ptimes.put(receiver.getMsgId(), previousAcc.getFirst());
			
			
			System.out.println("DiffMessage2: " + previousAcc.getSecond());
			psums.put(receiver.getMsgId(), previousAcc.getSecond());
			msg = contex.getCurrentMessage();
		}
		
		if (countMsg > 0) {
			
		accTims =   Utilties.fold(ptimes,otimes);
		accSum =  Utilties.fold(psums,oplus);

		//zipList(psums, firstEles, diff);

		System.out.println("step 3, foldl accTims" + accTims);

	//	accSum = (T2) Utilties.foldl(psums, oplus); // only for bsp
		System.out.println("step 3, foldl  accSum" + accSum);
	}
		
//
//		// do scan
//		// but each psum lacks the first element ( p( E(n) ,c * q E(n-1)
//		// ) )
//		if (countMsg > 0) {
//			accTims = otimes.evaluate(cons_c, (T1) Utilties.foldl(ptimes, otimes));
//
//			zipList(psums, firstEles, diff);
//
//			System.out.println("step two, zipList, psums" + psums);
//
//			accSum = (T2) Utilties.foldl(psums, oplus); // only for bsp
//			System.out.println("step two, foldl, accSum" + accSum);
//		} else {
//			// 1 st chunk
//			accTims = cons_c;
//		}
//
//		while (contex.readNext(key, line)) {
//			if (chunkey < 0) {
//				chunkey = key.get();
//				if (firstEles.floorEntry(chunkey) != null) {
//					lastOne = firstEles.get(firstEles.floorEntry(chunkey))
//							.getSecond();
//				}
//
//			} // init over
//
//			StringTokenizer itr = new StringTokenizer(line.toString().trim(),
//					DELIM);
//			StringBuffer outLine = new StringBuffer(1024);
//			T0 thisOne;
//			T1 lastAccTims = accTims;
//			while (itr.hasMoreTokens()) {
//				// sum += Integer.parseInt(itr.nextToken());
//				String tmp = itr.nextToken();
//
//				T2 result = oplus.id();
//
//				thisOne = insT0.parse(tmp);
//
//				if (chunkey == 0 && step == 0) { // first chunk , 1st step
//					T2 px0 = p.evaluate(thisOne, cons_c);
//					accSum = oplus.evaluate(accSum, px0);
//				}
//
//				if (lastOne != null && step == 0) {
//					// only do at the very beginning
//					T1 qLst = q.evaluate(lastOne);
//					T1 cqLst = otimes.evaluate(cons_c, qLst);
//					T2 pThiscqLst = p.evaluate(thisOne, cqLst);
//					accSum = oplus.evaluate(accSum, pThiscqLst);
//				}
//
//				T1 qThis = q.evaluate(thisOne);
//				// T2 rst = accSum + g.evaluate( otimes.evaluate(accTims, qThis)
//				// );
//
//				accTims = otimes.evaluate(accTims, qThis);
//
//				T2 pacc = p.evaluate(thisOne, lastAccTims);
//
//				accSum = oplus.evaluate(accSum, pacc);
//
//				if (itr.hasMoreTokens())
//					lastAccTims = accTims;
//				step++;
//			}
//		}
//
		T2 result = oplus.evaluate(accSum, g.evaluate(accTims));

		contex.write(new Text(result.toString()), NullWritable.get());
	}

	// in place update list1
	private void zipList(ScanDList<T2> list1,
			final ScanDList<Pair<T0, T0>> list2,
			DiffComputation<T0, T1, T2> diff) throws InterruptedException {
		long key = list1.firstKey(); // from first chunk-key, i.e., 0
		long higher = -1;
		while (list1.higherKey(key) != null) {
			higher = list1.higherKey(key);

			Pair<T0, T0> eFsteLst = list2.get(key);
			Pair<T0, T0> nextEFsteLst = list2.get(higher);

			T1 qx0 = diff.q.evaluate(eFsteLst.getSecond());
			T1 cqx0 = diff.otimes.evaluate(diff.init, qx0);
			T2 px1cqx0 = diff.p.evaluate(nextEFsteLst.getFirst(), cqx0);
			T2 fixed = diff.oplus.evaluate(px1cqx0, list1.get(key));
			list1.put(higher, fixed);
			key = higher;
		}
	}

}